I think this can be replaced with a map that the intercept function takes.

e.g.

#{protocol => mqtt,
    vhost => <<"/">>,
    client_id =><<"test">>}

i don't think we need to distinguish incoming and outgoing interceptors are the module / behaviour level.
We should have a single behaviour rabbit_message_interceptor with a single intercept function that takes the message container, the protocol context map (see another comment) and the config.

Instead they would be distinguined by which configuration they are added to. It is fine if this PR only implements incoming interceptors.

I think it would be better to use a tuple key for persistent term than doing this. e.g. {message_interceptors, incoming | outgoing}

This needs for ad hoc code to be put in rabbit:persist_static_configuration

    persist_static_configuration(
      [classic_queue_index_v2_segment_entry_count,
       classic_queue_store_v2_max_cache_size,
       classic_queue_store_v2_check_crc32,
       incoming_message_interceptors,
       outgoing_message_interceptors
      ]),

needs to become something like

    persist_static_configuration(
      [classic_queue_index_v2_segment_entry_count,
       classic_queue_store_v2_max_cache_size,
       classic_queue_store_v2_check_crc32,
       {incoming_message_interceptors, {message_interceptors, incoming}
       outgoing_message_interceptors, {message_interceptors, outgoing}
      ]),

and

persist_static_configuration(Params) ->
    lists:foreach(
      fun(Param) ->
              case application:get_env(?MODULE, Param) of
                  {ok, Value} ->
                      ok = persistent_term:put(Param, Value);
                  undefined ->
                      ok
              end
      end, Params).

needs to become

persist_static_configuration(Params) ->
    lists:foreach(
        fun({Param, Alias}) ->
            case application:get_env(?MODULE, Param) of
                {ok, Value} ->
                    ok = persistent_term:put(Alias, Value);
                undefined ->
                    ok
            end;
           (Param) ->
            case application:get_env(?MODULE, Param) of
                {ok, Value} ->
                    ok = persistent_term:put(Param, Value);
                undefined ->
                    ok
            end
        end, Params).

Is that fine? What about this other option:

The cuttlefish translations could generate a rabbit.message_interceptors.<incoming|outgoing> config instead of rabbit.<incoming_message_interceptors|outgoing_message_interceptors> and the rabbit:persist_static_configuration would only receive message_interceptors instead of both <incoming|outgoing>_message_interceptors. The interceptor module would read app conf. with message_interceptors and then then get either the incoming or the outgoing ones from the read term.

The app conf would look like

[{rabbit, [{message_interceptors, #{incoming => [...], outgoing => [...] ...

instead of

[{rabbit, [...
    {incoming_message_interceptors, ...},
    {outgoing_message_interceptors,...},
...

I see. that seems complicated. Perhaps just resolve incoming to incoming_message_interceptors etc in a function rather than doing atom to list conversions.

resolve(incoming) ->
   incoming_message_interceptors;
resolve(outgoing) ->
   outgoing_message_interceptors.

Yes, calling atom_to_list for each message that RabbitMQ receives creates lots of garbage, I wouldn't be surprised if performance drops by a few percent.

What Karl suggested is great.

Alternatively, it might be simpler to omit these translations entirely. For example instead of

-spec intercept(Msg, MsgInterceptorCtx, MsgDirection) -> Resp when
    Msg :: mc:state(),
    MsgInterceptorCtx :: map(),
    MsgDirection:: incoming | outgoing,
    Resp :: mc:state().

we could have

-spec intercept(Msg, Context, Group) -> Msg when
    Msg :: mc:state(),
    Context :: map(),
    Group :: incoming_message_interceptors | outgoing_message_interceptors.

I like the group option actually. I think it should be it's own argument to intercept/4 though so we don't have to modify the context map for each call.

I feel the direction should be passed to the interceptor, you may want to just have a single interceptor for both incoming and outgoing.

I assumed an interceptor would do the same independently of the direction of the message. It is the user the one that decides (if the interceptor allows it by providing the appropriate message_interceptors.<direction>. cuttlefish mapping) in which direction should the interceptor be applied, or if it should be applied in both.

As suggested by @gomoripeti , the rabbit_message_interceptor module could also add the message direction in the MsgInterceptorCtx map. How does that sound?

that sounds good.

Can we define some context properties that we'd always pass, e.g. vhost, username and the current protocol itself?

e.g.

#{protocol := amqp091 | amqp | mqttv3 | mqttv5, %% tags to be defined
   vhost := binary(),
   username := binary(),
   conn_name => binary() % optional
   atom() -> term()}

This should probably be x-opt-client-id or x-opt-mqtt-client-id or x-opt-mqtt-publisher-client-id such that this message annotation will be received by an AMQP 1.0 client in the message-annotation section.
The current annotation client_id also gets lost when the MQTT client publishes into a stream. Maybe this key should be a configuration option by the plugin.

x-opt-mqtt-client-id gets my vote.

yes, I suppose it's best to include mqtt so that it's clear that this is an MQTT client ID.

"Client ID" is otherwise a broad term which could for example refer to a JMS client ID (https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1#client-identifier) whose purpose is similar though:

The purpose of client identifier is to associate a connection and its objects with a state maintained on behalf of the client by a provider. By definition, the client state identified by a client identifier can be ‘in use’ by only one client at a time.

Can you add unit tests?
deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets is the file to add these tests.

this is great and what we discussed as we think lower memory overhead is preferrable. It would be good if it at all possible you could run some throughput tests so we can get a view on how much this might affect performance. https://www.rabbitmq.com/blog/2023/03/21/native-mqtt#latency-and-throughput

Run the benchmarks against the last commit (build on publish) and with the last one commit building the context map on init (build on init).

On an Arch Linux w/ 10 cores and 24 GB of RAM.

QOS1 build on publish

================= TOTAL PUBLISHER (100) =================
Total Publish Success Ratio:   100.000% (1000000/1000000)
Total Runtime (sec):           63.151
Average Runtime (sec):         62.670
Pub time min (ms):             0.134
Pub time max (ms):             39.215
Pub time mean mean (ms):       6.258
Pub time mean std (ms):        0.044
Average Bandwidth (msg/sec):   159.574
Total Bandwidth (msg/sec):     15957.402

================= TOTAL SUBSCRIBER (100) =================
Total Forward Success Ratio:      100.000% (1000000/1000000)
Forward latency min (ms):         0.126
Forward latency max (ms):         36.001
Forward latency mean std (ms):    0.020
Total Mean forward latency (ms):  3.128


QOS0 build on publish

================= TOTAL PUBLISHER (100) =================
Total Publish Success Ratio:   100.000% (1000000/1000000)
Total Runtime (sec):           1.352
Average Runtime (sec):         0.966
Pub time min (ms):             0.002
Pub time max (ms):             52.091
Pub time mean mean (ms):       0.036
Pub time mean std (ms):        0.014
Average Bandwidth (msg/sec):   14429.931
Total Bandwidth (msg/sec):     1442993.119

================= TOTAL SUBSCRIBER (100) =================
Total Forward Success Ratio:      75.687% (756868/1000000)
Forward latency min (ms):         7.968
Forward latency max (ms):         4167.565
Forward latency mean std (ms):    345.354
Total Mean forward latency (ms):  1803.073


QOS1 build on init

================= TOTAL PUBLISHER (100) =================
Total Publish Success Ratio:   100.000% (1000000/1000000)
Total Runtime (sec):           63.087
Average Runtime (sec):         62.568
Pub time min (ms):             0.142
Pub time max (ms):             31.975
Pub time mean mean (ms):       6.248
Pub time mean std (ms):        0.054
Average Bandwidth (msg/sec):   159.839
Total Bandwidth (msg/sec):     15983.907

================= TOTAL SUBSCRIBER (100) =================
Total Forward Success Ratio:      100.000% (1000000/1000000)
Forward latency min (ms):         0.138
Forward latency max (ms):         19.824
Forward latency mean std (ms):    0.025
Total Mean forward latency (ms):  3.112


QOS0 build on init

================= TOTAL PUBLISHER (100) =================
Total Publish Success Ratio:   100.000% (1000000/1000000)
Total Runtime (sec):           1.423
Average Runtime (sec):         0.961
Pub time min (ms):             0.002
Pub time max (ms):             38.604
Pub time mean mean (ms):       0.030
Pub time mean std (ms):        0.008
Average Bandwidth (msg/sec):   12087.794
Total Bandwidth (msg/sec):     1208779.411

================= TOTAL SUBSCRIBER (100) =================
Total Forward Success Ratio:      75.167% (751674/1000000)
Forward latency min (ms):         1.671
Forward latency max (ms):         4043.116
Forward latency mean std (ms):    207.787
Total Mean forward latency (ms):  1765.429

I get that the under-100% Total Forward Success Ratio on subscribers for QOS 0 comes from me using not a boxed environment + reduced RAM compared to the one used on the doc page you linked.

So unexpectedly the qos0 test is faster creating the map dynamically?

It seems it is, but does not really make too much sense to me. I'll repeat this benchmark on a dedicated machine just in case after addressing the squashed commit writing.

It seems this file is untested?

Adding the following simple test would be good enough:

• MQTT client publishes a single message.
• Different clients (MQTT, AMQP 1.0, AMQP 0.9.1) receive this message validating that it contains the MQTT client ID of the publisher.

... let me know if you have difficulties adding such a test case (I can add it as well)

As you can see from my last commit, I'm struggling to get the published message with the amqp10_client. If you want to finish that up, I won't come back to this until tomorrow ~0830.

Ok, I will add a test.

I don't think this addition is necessary because rabbitmq_amqp_client already depends on amqp10_client.